Skip to content

#1390 - configurable storage for aggregating functions#2473

Open
MrHDOLEK wants to merge 10 commits into
flow-php:1.xfrom
MrHDOLEK:#1390
Open

#1390 - configurable storage for aggregating functions#2473
MrHDOLEK wants to merge 10 commits into
flow-php:1.xfrom
MrHDOLEK:#1390

Conversation

@MrHDOLEK

@MrHDOLEK MrHDOLEK commented Jun 21, 2026

Copy link
Copy Markdown
Contributor

Makes GroupBy aggregation memory-efficient and configurable (issue #1390): keep the fast in-memory path by default, or opt into an external, bounded-memory path.

The external path is sort-then-fold: sort the rows by the group-by columns (reusing ExternalSort, which already spills to disk and merges in bounded memory), then fold adjacent equal keys in a single pass. Because a group is never split across independent partial accumulators, there is no merge contract — the whole thing is just "sort, then collapse equal keys", mirroring how ExternalSort works. Output is streamed in batches instead of one big Rows.

aggregationAlgorithm(AggregationAlgorithms::MEMORY_AGGREGATION | EXTERNAL_AGGREGATION) selects the strategy (mirrors SortConfigBuilder::algorithm()); the default stays in-memory, so existing pipelines are unchanged.

Addressing the review

  • GroupEntry "keeps values for all aggregators?" — gone. There is no storage layer; the algorithms hold their accumulators directly.
  • aggregationStorage(strategy) — "storage or strategy?" and aggregationStorage() vs aggregationStore() — replaced by a single aggregationAlgorithm(AggregationAlgorithms); the "storage/store" naming is removed.
  • AggregationStorage::entry() vs save() — gone with the storage interface; aggregation is one inline loop.
  • MergeableAggregatingFunction "when can two functions merge?" — removed entirely. Sort-fold never merges partial aggregates, so the interface and merge() on the built-ins are deleted.

Correctness note: the external path tags each row with its input position and sorts by (group keys, input ordinal), so first() / last() / unsorted collect() return the same result as the in-memory path even when a group spills across sort runs (covered by a test).

Resolves: #1390

Change Log


Added

  • Configurable, memory-efficient GroupBy aggregation: ConfigBuilder::aggregationAlgorithm() and the AggregationAlgorithms enum select between MEMORY_AGGREGATION (default, in-memory hash map) and EXTERNAL_AGGREGATION (sorts by the group-by columns and folds, spilling to disk for bounded memory), with aggregationFilesystem() for the spill protocol.
  • GroupBy streams its result in batches instead of materializing every group into a single Rows.

Fixed

Changed

Removed

Deprecated

Security

GroupBy aggregation can move state out of the PHP heap instead of holding every result in memory. A new AggregationStorage abstraction offers three backends: memory (default, unchanged behavior), filesystem spill+merge, and PSR-16 key-value (APCu/Redis).

- MergeableAggregatingFunction + merge() on all built-in aggregators
- AggregationStorage: Memory / External (filesystem spill+merge) / Kv (PSR-16)
- KV storage fails loudly on eviction / write failure instead of returning a wrong aggregate
- config: AggregationStorageStrategy + FLOW_AGGREGATION_MAX_MEMORY wired into ConfigBuilder
- APCu enabled in the nix dev shell (for the PSR-16/APCu storage test)
@codecov

codecov Bot commented Jun 21, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 92.46575% with 22 lines in your changes missing coverage. Please review.
✅ Project coverage is 85.99%. Comparing base (220cf27) to head (69df77e).

Additional details and impacted files
@@             Coverage Diff              @@
##                1.x    #2473      +/-   ##
============================================
+ Coverage     85.97%   85.99%   +0.02%     
- Complexity        0    22337   +22337     
============================================
  Files          1662     1670       +8     
  Lines         68370    68583     +213     
============================================
+ Hits          58783    58981     +198     
- Misses         9587     9602      +15     
Components Coverage Δ
etl 88.68% <92.46%> (+0.06%) ⬆️
cli 89.40% <ø> (ø)
lib-array-dot 81.44% <ø> (ø)
lib-azure-sdk 64.44% <ø> (ø)
lib-doctrine-dbal-bulk 93.61% <ø> (ø)
lib-filesystem 85.97% <ø> (ø)
lib-types 90.18% <ø> (ø)
lib-parquet 70.10% <ø> (ø)
lib-parquet-viewer 82.26% <ø> (ø)
lib-snappy 90.26% <ø> (+0.88%) ⬆️
lib-dremel 0.00% <ø> (ø)
lib-postgresql 88.62% <ø> (ø)
lib-telemetry 86.58% <ø> (ø)
bridge-filesystem-async-aws 92.74% <ø> (ø)
bridge-filesystem-azure 90.45% <ø> (ø)
bridge-monolog-http 96.82% <ø> (ø)
bridge-monolog-telemetry 94.79% <ø> (ø)
bridge-openapi-specification 92.07% <ø> (ø)
symfony-http-foundation 78.57% <ø> (ø)
bridge-psr18-telemetry 100.00% <ø> (ø)
bridge-psr3-telemetry 98.95% <ø> (ø)
bridge-psr7-telemetry 100.00% <ø> (ø)
bridge-telemetry-otlp 90.41% <ø> (ø)
bridge-symfony-http-foundation-telemetry 92.85% <ø> (ø)
bridge-symfony-filesystem-bundle 90.66% <ø> (ø)
bridge-symfony-filesystem-cache 98.18% <ø> (ø)
bridge-symfony-postgresql-bundle 93.39% <ø> (ø)
bridge-symfony-postgresql-cache 94.41% <ø> (ø)
bridge-symfony-postgresql-messenger 98.80% <ø> (ø)
bridge-symfony-postgresql-session 93.65% <ø> (ø)
bridge-symfony-telemetry-bundle 90.23% <ø> (ø)
adapter-chartjs 84.05% <ø> (ø)
adapter-csv 91.43% <ø> (ø)
adapter-doctrine 90.79% <ø> (ø)
adapter-google-sheet 99.18% <ø> (ø)
adapter-http 72.34% <ø> (ø)
adapter-json 88.71% <ø> (ø)
adapter-logger 50.00% <ø> (ø)
adapter-parquet 77.70% <ø> (ø)
adapter-text 74.13% <ø> (ø)
adapter-xml 83.40% <ø> (ø)
adapter-avro 0.00% <ø> (ø)
adapter-excel 94.21% <ø> (ø)
adapter-postgresql 91.06% <ø> (ø)
adapter-seal 85.42% <ø> (ø)
bridge-phpunit-postgresql 75.30% <ø> (ø)
bridge-phpunit-telemetry 87.32% <ø> (ø)
bridge-phpstan-types 0.00% <ø> (ø)
bridge-postgresql-valinor 100.00% <ø> (ø)
🚀 New features to boost your workflow:
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@datadog-official

This comment has been minimized.

*/
public function __construct(
public array $values,
public array $aggregators,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not really sure if I understand this data structure, it's keeping values for all aggregators?

return $this;
}

public function aggregationStorage(AggregationStorageStrategy $strategy): self

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

so storage or strategy? 😅

return $this;
}

public function aggregationStore(AggregationStorage $storage): self

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Those two methods, aggregationStorage and aggregationStore are a bit confusing


public function flush(): void;

public function save(string $hash, GroupEntry $entry): void;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what's the difference between entry and save?


namespace Flow\ETL\Function;

interface MergeableAggregatingFunction extends AggregatingFunction

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could you walk me through this logic? Like when two AggregatingFunctions can be merged?

MrHDOLEK added 3 commits July 6, 2026 22:10
…torage

Reworks the aggregation-storage approach following review. External aggregation is now sort-then-fold: sort rows by the group-by columns (reusing ExternalSort, which spills and merges in bounded memory) and fold adjacent equal keys. This makes the whole abstraction simpler and drops the concepts the review found confusing.

- remove the AggregationStorage layer (Memory/External/Kv storage, GroupEntry, GroupBucket(sCache)) and the AggregationStorageStrategy enum
- remove MergeableAggregatingFunction and merge() from aggregators - sort-fold never splits a group, so no merge contract is needed
- add AggregatingAlgorithm with MemoryAggregation (default, hash map) and ExternalAggregation (sort + fold), selected via AggregationAlgorithms (mirrors SortAlgorithms)
- GroupBy is now a spec (references/aggregations/hash/keyValues); GroupByProcessor routes pivot and global aggregation to memory
- config: aggregationAlgorithm() replaces aggregationStorage()/aggregationStore(); output is streamed in batches instead of one Rows
- external tags rows with an input ordinal and sorts by (group keys, ordinal), so first()/last()/collect() return the same result as the in-memory path even when a group spills across sort runs
@github-actions github-actions Bot added size: L and removed size: XL labels Jul 6, 2026
@MrHDOLEK MrHDOLEK requested a review from norberttech July 6, 2026 20:26
Non-pivot aggregation now runs through hash aggregation with adaptive spilling: groups accumulate in an in-memory hash map, and once the configured memory limit is exceeded the map is spilled to a hash-sorted bucket and cleared. If nothing spills the map is the result; otherwise a streaming k-way merge folds partial aggregates by group key, so spilled state is O(groups), not O(rows).

- merge() moves onto the base AggregatingFunction contract (no separate opt-in interface); every built-in aggregator implements it and self-validates by reference and result-affecting arguments
- HashAggregation + GroupState / BucketGroup / GroupsMinHeap / GroupBucketsCache (FilesystemGroupBucketsCache), mirroring the external-sort building blocks
- GroupBy becomes a spec (references/aggregations/hash/keyValues/newAggregators); pivot stays memory-only
- config: aggregationMemoryLimit(Unit) (default unbounded, opt-in spilling) replaces the algorithm enum; output streamed in batches
- fix multi-column group-by key hash collision (length-prefix the values)
@github-actions github-actions Bot added size: XL and removed size: L labels Jul 7, 2026
MrHDOLEK added 4 commits July 7, 2026 08:43
- guard strpos()===false and drop unused mago pragmas flagged by CI static analysis
- annotate the bucket-cursor generators as Generator<int, array{...}> and narrow current() in the k-way merge
- cover merge() on all 10 aggregators (success, null branches, self-validation), the min-heap ordering/guard, the bucket cache round-trip, and the memory-limit env var
- drop the unused mago pragma in Sum::merge
- hoist the pivot prototype behind an instanceof guard so clone is never applied to a possibly-false current()
- drop over-specific @return docblocks that toArray() cannot satisfy
private function hotCursor(array $hot): Generator
{
foreach ($hot as $hash => $state) {
yield [$hash, $state];

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please don't use tuples for return types, introduce a simple VO instead

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Storage for Aggregating Functions

2 participants